<?php
error_reporting(0);
include_once("interface_concierge.php");
include_once("class_callback.php");

class Concierge implements ConciergeInterface {
	private $mysqli;
	private $callback_handler;

	public function __construct(&$mysqli) {
		$this->mysqli = $mysqli;
		$this->callback_handler = new Callback();
	}
	
	private function idForPrincipal($name) {
		$stmt = $this->mysqli->prepare("select `pid` from `principals` where `name` = ?");
		$stmt->bind_param("s",$name);
		$stmt->execute();
		$stmt->store_result();
		
		if($stmt->errno != 0 OR $stmt->num_rows != 1) {
			$stmt->close();
			return FALSE;
		}

		$stmt->bind_result($pid);
		$stmt->fetch();
		$stmt->close();

		return $pid;
	}
	
	
	public function registerPrincipal($name, $key, $callback_url) {
		$stmt = $this->mysqli->prepare("insert into `principals` (`name`, `key_hash`, `callback`, `key_salt`) values (?, SHA1(CONCAT(?, SHA1(NOW()))), ?, SHA1(NOW()))");

		$stmt->bind_param("sss",$name,$key,$callback_url);
		$stmt->execute();

		if($stmt->errno != 0 OR $stmt->affected_rows != 1) {
			$stmt->close();
			return FALSE;
		}
		
		$insert_id = $stmt->insert_id;
		$stmt->close();
				
		return $insert_id;
	}
	
	public function unregisterPrincipal($name) {	
		$stmt = $this->mysqli->prepare("delete from `principals` where `name` = ?");
		$stmt->bind_param("s",$name);
		$stmt->execute();
		
		if($stmt->errno != 0 OR $stmt->affected_rows != 1) {
			$stmt->close();
			return FALSE;
		}
		
		$stmt->close();
		return TRUE;
	}
	
	public function authenticatePrincipal($name, $key) {
		$stmt = $this->mysqli->prepare("select `pid` from `principals` where `name` = ? and `key_hash` = SHA1(CONCAT(?, `key_salt`))");
		$stmt->bind_param("ss", $name, $key);
		$stmt->execute();
		$stmt->store_result();
		
		if($stmt->errno != 0 OR $stmt->num_rows != 1) {
			$stmt->close();
			return FALSE;
		}

		$stmt->bind_result($pid);
		$stmt->fetch();
		$stmt->close();

		return $pid;
	}
	

	
	private function serializeTagArray(array $sorted_tag_array) {
		$output = "";
		$first = TRUE;
		foreach ($sorted_tag_array as $tag) {
			if(!$first) $output .= ", ";
			else $first = FALSE;
			$output .= $tag;
		}
		return $output;
	}
	
	public function registerTagInterest($name, array $tags) {
		$stmt = $this->mysqli->prepare("insert into `conjuncts` (`pid`,`taglist_hash`) values((select `pid` from `principals` where `name` = ? limit 1), ?)");
		
		sort($tags);
		$conjunct_hash = sha1($this->serializeTagArray($tags));
		
		$stmt->bind_param("ss", $name, $conjunct_hash);
		$stmt->execute();

		if($stmt->errno != 0) {
			$stmt->close();
			return FALSE;
		}

		$insert_id = $stmt->insert_id;
		$stmt->close();

		
		$this->insertConjunctTags($insert_id, $tags);
		
		return TRUE;
	}
	
	private function insertConjunctTags($cid, array $tags) {
		$stmt = $this->mysqli->prepare("insert into `conjunct_tags` (`cid`, `tag`) values (?, ?)");
		
		$tag = NULL;
		$stmt->bind_param("ds", $cid, $tag);
		
		foreach ($tags as &$the_tag) {
			$tag = $the_tag;
			$stmt->execute();
		}
		$stmt->close();
	}
	
	public function unregisterTagInterest($name, array $tags) {
		$stmt = $this->mysqli->prepare("delete from `conjuncts` where `pid` = (select `pid` from `principals` where `name` = ? limit 1) and `taglist_hash` = ?");
	
		sort($tags);
		$conjunct_hash = sha1($this->serializeTagArray($tags));
		
		$stmt->bind_param("ss", $name, $conjunct_hash);
		$stmt->execute();
		
		if($stmt->errno != 0) {
			$stmt->close();
			return FALSE;
		}

		$stmt->close();
		
		return TRUE;
	}
	
	public function unregisterAllTagInterests($name) {
		$stmt = $this->mysqli->prepare("delete from `conjuncts` where `pid` = (select `pid` from `principals` where `name` = ? limit 1)");
		
		$stmt->bind_param("s", $name);
		$stmt->execute();
		
		if($stmt->errno != 0) {
			$stmt->close();
			return FALSE;
		}
		
		$stmt->close();
		
		return TRUE;
	}
	
	private function &getTagsForConjunct($cid) {
		$result = array();
		$stmt = $this->mysqli->prepare("select `tag` from `conjunct_tags` where `cid`= ?");
		
		$stmt->bind_param("d", $cid);
		$stmt->execute();

		if($stmt->errno != 0) {
			$stmt->close();
			return FALSE;
		}
		
		$stmt->bind_result($tag);
		
		while($stmt->fetch()) {
			$result[] = $tag;
		}

		return $result;
	}
	
	public function getTagInterests($name) {
		$result = array();
	
		$stmt = $this->mysqli->prepare("select `cid` from `conjuncts` where `pid` = (select `pid` from `principals` where `name` = ? limit 1)");
		$stmt->bind_param("s", $name);
		$stmt->execute();
	
		if($stmt->errno != 0) {
			$stmt->close();
			return FALSE;
		}
		
	
		$stmt->bind_result($cid);
		$cids = array();
		while($stmt->fetch()) {
			$cids[] = $cid;
		}
		
		$stmt->close();
		
		foreach($cids as &$mcid) {
			$result[] = $this->getTagsForConjunct($mcid);
		}


		return $result;
	}
	
	private function setGlobalInterest($name, $value) {
		$stmt = $this->mysqli->prepare("update `principals` set `interested_all` = ? where `name` = ?");
		$stmt->bind_param("ds", $value, $name);
		$stmt->execute();
		
		if($stmt->errno != 0 /* OR $stmt->affected_rows != 1*/) {
			$stmt->close();
			return FALSE;
		}

		$stmt->close();
		return TRUE;		
	}
	
	public function registerGlobalInterest($name) {
		return $this->setGlobalInterest($name, 1);
	}
	
	public function unregisterGlobalInterest($name) {
		return $this->setGlobalInterest($name, 0);
	}
	
	public function registerPrincipalInterest($name, $other) {
		$stmt = $this->mysqli->prepare("insert into `principal_follow` (`follower_pid`, `followed_pid`) values ((select `pid` from `principals` where `name` = ? limit 1), (select `pid` from `principals` where `name` = ? limit 1))");
		$stmt->bind_param("ss", $name, $other);
		$stmt->execute();

		if($stmt->errno != 0 OR $stmt->affected_rows != 1) {
			$stmt->close();
			return FALSE;
		}
		
		$stmt->close();		
		return TRUE;
	}
	
	private function getPrincipalInterested($name) {
		$stmt = $this->mysqli->prepare("select `follower_pid` from `principal_follow`, `principals` where `name` = ? and `followed_pid` = `pid`");
		$stmt->bind_param("s", $name);
		$stmt->execute();
	
		if($stmt->errno != 0) {
			$stmt->close();
			return FALSE;
		}
		
		$interests = array();
		
		$stmt->bind_result($pid);
		
		while($stmt->fetch()) {
			$interests[] = array("pid" => $pid);
		}
		
		return $interests;
	}
	
	
	public function getPrincipalInterests($name) {
		$stmt = $this->mysqli->prepare("select `followed_pid`, `name` from `principal_follow`, `principals` where `follower_pid` = (select `pid` from `principals` where `name` = ? limit 1) and `followed_pid` = `pid`");
		$stmt->bind_param("s", $name);
		$stmt->execute();
	
		if($stmt->errno != 0) {
			$stmt->close();
			return FALSE;
		}
		
		$interests = array();
		
		$stmt->bind_result($pid, $iname);
		
		while($stmt->fetch()) {
			$interests[] = array("pid" => $pid, "name" => $iname);
		}
		
		return $interests;
	}
	
	public function unregisterPrincipalInterest($name, $other) {
		$stmt = $this->mysqli->prepare("delete from `principal_follow` where `follower_pid` = (select `pid` from `principals` where `name` = ? limit 1) and `followed_pid` = (select `pid` from `principals` where `name` = ? limit 1)");
		$stmt->bind_param("ss", $name, $other);
		$stmt->execute();

		if($stmt->errno != 0 OR $stmt->affected_rows != 1) {
			$stmt->close();
			return FALSE;
		}
		
		$stmt->close();		
		return TRUE;
	}
	
	public function unregisterAllPrincipalInterests($name) {
		$my_pid = $this->idForPrincipal($name);
		
		$stmt = $this->mysqli->prepare("delete from `principal_follow` where `follower_pid` = (select `pid` from `principals` where `name` = ? limit 1)");
		$stmt->bind_param("s", $name);
		$stmt->execute();

		if($stmt->errno != 0) {
			$stmt->close();
			return FALSE;
		}
		
		$stmt->close();		
		return TRUE;
	}
	

	
	public function unregisterAllInterests($name) {
		return ($this->unregisterAllPrincipalInterests($name) AND
					$this->unregisterGlobalInterest($name) AND
					$this->unregisterAllTagInterests($name));
	}

	private function insertMessageTags($mid, array $tags) {
		$stmt = $this->mysqli->prepare("insert into `message_tags` (`mid`, `tag`) values (?, ?)");
		
		$tag = NULL;
		$stmt->bind_param("ds", $mid, $tag);
		
		foreach ($tags as $the_tag) {
			$tag = $the_tag;
			$stmt->execute();
		}
		$stmt->close();
	}
	
	
	private function refValues($arr) { 
		if (strnatcmp(phpversion(),'5.3') >= 0) //Reference is required for PHP 5.3+ 
		{ 
			$refs = array(); 
			foreach($arr as $key => $value) 
				$refs[$key] = &$arr[$key]; 
			return $refs; 
		} 
		return $arr; 
	
	} 
	
	private function PIDforCID($cid) {
		$stmt = $this->mysqli->prepare("select `pid` from `conjuncts` where `cid` = ?");
		$stmt->bind_param("d",$cid);
		$stmt->execute();
		$stmt->store_result();
		
		if($stmt->errno != 0 OR $stmt->num_rows != 1) {
			$stmt->close();
			return FALSE;
		}

		$stmt->bind_result($pid);
		$stmt->fetch();
		$stmt->close();

		return $pid;
	}
	
	private function addPIDifCIDmatches($cid, array &$pids, array &$tags) {
		$conjunct_tags = $this->getTagsForConjunct($cid);
		
		$diff = array_diff($conjunct_tags, $tags);
		
		if(count($diff) == 0) {
			$pids[] = $this->PIDforCID($cid);
		}
	}
	
	private function &getTaglistInterested(array $tags) {
		$count  = count($tags);
		$bind_config = "";
		$query_values = "";
		for($i = 0; $i < $count; $i++) {
			$bind_config .= "s";
			if($i != 0) $query_values .= ", ";
			$query_values .= "?";
		}
		
		$args = array_merge(array($bind_config), $tags);
		$query = "select distinct `cid` from `conjunct_tags` where `tag` in (".$query_values.")";
				
		$stmt = $this->mysqli->prepare($query);
		call_user_func_array(array($stmt, "bind_param"), $this->refValues($args)); 

		$interested_pids = array();
		
		$stmt->execute();
		
		if($stmt->errno != 0) {
			$stmt->close();
			return FALSE;
		}
		
		$stmt->bind_result($cid);
		
		$cids = array();
		while($stmt->fetch()) {
			$cids[] = $cid;
		}
		
		$stmt->close();

		
		foreach($cids as &$mcid) {
			$this->addPIDifCIDmatches($mcid, $interested_pids, $tags);
		}
		
		
		return $interested_pids;
	}
	
	public function postMessage($sender_name, $content, array $tags) {
		$stmt = $this->mysqli->prepare("insert into `messages` (`snd_pid`, `content`, `taglist`, `snd_name`) values ((select `pid` from `principals` where `name` = ? limit 1), ?, ?, ?)");
		
		sort($tags);
		$serialized_tags = $this->serializeTagArray($tags);
		$stmt->bind_param("ssss", $sender_name, $content, $serialized_tags, $sender_name);
		$stmt->execute();

		if($stmt->errno != 0 OR $stmt->affected_rows != 1) {
			//echo $stmt->error;
			$stmt->close();
			return FALSE;
		}

		$insert_id = $stmt->insert_id;
		$stmt->close();

		if(!$this->insertMessageTags($insert_id, $tags)) {
			// TODO handle failed transaction
		}
		
		
		$interested_pids = array();
		
		/***** Find principals with global interest *****/
		
		$stmt = $this->mysqli->prepare("select `pid` from `principals` where `interested_all` = 1");
		$stmt->execute();
		if($stmt->errno != 0) {
			$stmt->close();
			// TODO handle failed transaction
			return FALSE;
		}
		$stmt->bind_result($pid);
		while($stmt->fetch()) {
			$interested_pids[] = $pid;
		}
		
		/***** Find principals interested in the sending principal *****/
		
		$principal_interested = $this->getPrincipalInterested($sender_name);
		foreach($principal_interested as $principal_interest) {
			$interested_pids[] = $principal_interest['pid'];
		}
		
		/***** Find principals interested in the tags of the message *****/
		
		
		$principal_interested = $this->getTaglistInterested($tags);
		foreach($principal_interested as $principal_interest) {
			$interested_pids[] = $principal_interest;
		}
		
		/***** Find principals this message is directed to *****/
		
		foreach($tags as $tag){
			preg_match('/to\:(.*)/', $tag, $recipient_id);
			if(sizeof($recipient_id > 0)){
				$interested_pids[] = $recipient_id[1];
			}
		}
		

		$interested_pids = array_unique($interested_pids);
		
		$this->insertMessageIntoPrincipalQueues($insert_id, $interested_pids);
		$this->sendPrincipalCallbacks($insert_id, $interested_pids);
		
		return $insert_id;
	}
	
	private function sendPrincipalCallbacks($mid, array &$pids) {
		$stmt = $this->mysqli->prepare("select `callback` from `principals` where `pid` = ?");
		$pid = NULL;
		$stmt->bind_param("d", $pid);
		$stmt->bind_result($callback);

		foreach ($pids as $the_pid) {
			$pid = $the_pid;
			$stmt->execute();
			$stmt->fetch();
			$this->callback_handler->enqueueCallbackNonblocking($callback, $mid);
		}
		
		$stmt->close();
	}
	
	private function insertMessageIntoPrincipalQueues($mid, array &$pids) {
		$stmt = $this->mysqli->prepare("insert into `message_queue` (`pid`, `mid`) values (?, ?)");
		
		$pid = NULL;
		$stmt->bind_param("dd", $pid, $mid);
		
		foreach ($pids as $the_pid) {
			$pid = $the_pid;
			$stmt->execute();
		}
		
		$stmt->close();
	}
	
	
	public function getPrincipalInterestStream($name, $since = 0, $limit = 1000) {
		return $this->getAStream("principalInterest", $since, $name, $limit, NULL);
	}
	
	public function getPrincipalStream($name, $since = 0, $limit = 1000) {
		return $this->getAStream("principal", $since, $name, $limit, NULL);
	}

	public function getGlobalStream($since = 0, $limit = 1000) {
		return $this->getAStream("global", $since, NULL, $limit, NULL);
	}
	
	public function getTaglistStream(array $tags, $since = 0, $limit = 1000) {
		return $this->getAStream("tag", $since, NULL, $limit, $tags);
	}
	
	private function getAStream($which_stream = "global", $since = 0, $name = NULL, $limit = 1000, array $tags = NULL) {
		$output = array();
		$query_head = "select `mid`, `snd_pid`, `re_mid`, `content`, `taglist`, UNIX_TIMESTAMP(`timestamp`), `snd_name` from `messages` where ";
		$query_tail = " mid > ?  order by mid desc limit ? ";
		$stmt = NULL;

		switch($which_stream) {				
			case "principalInterest":
				$stmt = $this->mysqli->prepare($query_head . " `mid` in (select `mid` from `message_queue` where `pid` = (select `pid` from `principals` where `name` = ? limit 1) and `mid` > ?) and " . $query_tail);
				$stmt->bind_param("sddd", $name, $since, $since, $limit);
				break;	
				
			case "principal":
				$stmt = $this->mysqli->prepare($query_head . " `snd_pid` = (select `pid` from `principals` where `name` = ? limit 1) and " . $query_tail);
				$stmt->bind_param("sdd", $name, $since, $limit);
				break;
				
			case "tag":
				$tag_count = count($tags);
				if($tag_count > 0) {
					
					$bind_config = "";
					$query_values = "";
					for($i = 0; $i < $tag_count; $i++) {
						$bind_config .= "s";
						if($i != 0) $query_values .= ", ";
						$query_values .= "?";
					}
					$args = array_merge(array($bind_config."dddd"), $tags, array($since, $tag_count, $since, $limit));
					$query = $query_head . " `mid` in (SELECT `mid` FROM `message_tags` WHERE `tag` IN (".$query_values.") AND mid > ? GROUP BY `mid` HAVING COUNT(`tag`) = ?) and " . $query_tail;
					$stmt = $this->mysqli->prepare($query);
					call_user_func_array(array($stmt, "bind_param"), $this->refValues($args)); 
					
					break;
				}
				// no break here intentionally! do not insert a break!
				
				
			case "global":
			default:
				$stmt = $this->mysqli->prepare($query_head . $query_tail);
				$stmt->bind_param("dd", $since, $limit);
		
				break;
		}
		
	
		$stmt->execute();
		$stmt->bind_result($mid, $snd_pid, $re_mid, $content, $taglist, $timestamp, $snd_name); 

    	/* fetch values */ 
    	while ($stmt->fetch()) { 
        	$output[] = array(	"mid" => $mid,
        						"snd_pid" => $snd_pid,
        						"re_mid" => $re_mid,
        						"content" => $content,
        						"taglist" => $taglist,  // TODO: would it be useful to get an array of actual tags for this?
        						"timestamp" => $timestamp,
        						"snd_name" => $snd_name
        					 );
        }
        
        return $output;		
	
	}

}


?>
